/*!
并行计算模块

 * 使用python开发具体的算子模块（借助libloading技术的实现因为core dump而放弃）
 * 借用gfs的分布式能力进行算子的部署
 * 本模块在完成上述算子的部署后，基于router发布任务调度指令，各node服务器载入算子，借助rayon对目录下文件进行并发计算
 * 执行进度将被写入指定gfs writer，具体结果，python算子自行发挥

*/
use crate::*;
use anyhow::anyhow;
use axum::response::{IntoResponse, Response};
use gfsc::ParallelArgs;
use std::collections:: HashMap;
use std::process::Command;
// use itertools::Itertools;
use axum::body::Bytes;

use rayon::iter::{IntoParallelIterator, ParallelIterator};
use regex::Regex;
use serde_json::{json, Value};
use stat::Stat;
use tokio::sync::mpsc;
use tokio::time::Duration;
use log::Level::{Error,Info, Debug};

#[cfg(feature = "rotatelog")]
use log::{debug, error, info, warn, log as event};
#[cfg(feature = "tracelog")]
use tracing::{error,  event};


/*
动态载入算子的实现库，并借助rayon进行文件级别的并行处理。
 - 必须满足算子的3个函数规范。
   * fn file_pattern_filter(fname:&str) -> bool
   * fn proccess_file(fname:&str, args:bytes::Bytes)-> String
   * fn complete_notify(server: &str)
 - args实际是通过json数据传递的参数。
 - 返回的是 proccess_file的 String集合的jion("\n")结果。

后来发现频繁调用过程中经常出现不可控的core dump，只能放弃。
pub(crate) fn call_operator_dynamic(
    lib_path_name: &str,
    args: &bytes::Bytes,
    files: Vec<Stat>,
    parent: &PathBuf,
) -> Result<String> {
    let mut ret = String::new();
    unsafe {
        let lib = libloading::Library::new(lib_path_name)?;
        let proccess_file: libloading::Symbol<
            unsafe extern "C" fn(src: &str, args: &bytes::Bytes) -> String,
        > = lib.get(b"proccess_file")?;
        let file_filter: libloading::Symbol<unsafe extern "C" fn(src: &str) -> bool> =
            lib.get(b"file_pattern_filter")?;

        let re: Vec<_> = files
            .into_par_iter()
            .filter(|stat| file_filter(&stat.name))
            .map(|stat| {
                let fname = parent
                    .join(stat.fhash)
                    .join(format!("{:02x}", stat.id))
                    .join(&stat.name);
                fname.display().to_string()
            })
            .map(|filename| proccess_file(&filename, args))
            .collect();
        ret.push_str(&re.join("\n"));
    }
    Ok(ret)
}
 */

pub(crate) struct Parallel {
    pub shutdown: shutdown::Shutdown,
    /// Not used directly. Instead, when `Handler` is dropped...?
    pub _shutdown_complete: mpsc::Sender<()>,
    pub shared: Arc<ShareComm>,
}

impl Parallel {
    pub async fn run(&mut self) -> Result<()> {
        let mut rx = self.shared.prx.lock().await;

        let mut ticker = tokio::time::interval(Duration::from_secs(60 * 60));
        ticker.tick().await; // 首次立刻执行，不合适。

        event!(target:"startup", loglevel(Info), "Parallel thread is startup, OK");
        while !self.shutdown.is_shutdown() {
            tokio::select! {
                _ = ticker.tick() => {
                    if self.shared.is_master(){
                        let mut taskcache = self.shared.record.write().await;
                        taskcache.cache_check();
                    }
                },
                cmd = rx.recv() => {
                    if let Some(mut cmd) = cmd {
                        if let Err(err) = cmd.run( &self.shared).await {
                            error!("error to dispach task : {:#?}", err);
                        }
                    }
                },
                _ = self.shutdown.recv() => {
                    // If a shutdown signal is received, return from `run`.
                    // This will result in the task terminating.
                    //return Ok(());
                    break
                }
            }
        }
        event!(target:"shutdown", loglevel(Error), "Parallel thread is shutdown, OK");
        Ok(())
    }
}

/**
执行算子的预先条件，遵从正则，只要有匹配就算通过
 */
fn parallel_task_filter(pattern: &str, fname: &str) -> bool {
    event!(target: "parallel", loglevel(Debug), "pattern = {}, fname = {}", pattern, fname );

    if let Ok(regex) = Regex::new(pattern) {
        if let Some(rmatch) = regex.find(&fname) {
            event!(target: "parallel", loglevel(Debug), "regex filter: = {:#?}", rmatch);
            // println!("regex filter: = {:#?}", rmatch);
            return true;
        }
    } else {
        error!("error to Regex::new");
    }
    false
}
/**
执行算子，注意参数顺序： python3 operator.py obj.log args-json
 */
fn operator_script_excute(script: &str, fname: &str, args: &str) -> bool {
    event!(target: "parallel", loglevel(Debug), "script = {}, fname = {}", script, fname );
    // std::time::sleep(Duration::from_secs(1));
    let args_encode = urlencoding::encode(&args);
    let file_encode = urlencoding::encode(&fname);
    if let Ok(out) = Command::new("python3")
        .arg(script)
        .arg(file_encode.as_ref())
        .arg(args_encode.as_ref())
        .output()
    {
        let s = String::from_utf8_lossy(&out.stderr);
        if s.len() > 0 {
            error!("{}", s);
            return false;
        }
        println!("OK, {:#?}", out);
        return true;
    }
    false
}

/**
执行算子，借助rayon进行文件级别的并行处理。
 - 算子的函数规范。
    * 算子是由python实现
    * 传给算子的第一参数是目标文件，第二参数是urlencode后的json数据，是参数集合

*/

pub(crate) fn call_operator_script(
    lib_path_name: &str,
    args: &str,
    files: Vec<Stat>,
    parent: &PathBuf,
) -> Result<()> {
    // let mut ret = String::new();
    files
        .into_par_iter()
        .filter(|stat| parallel_task_filter("(.*).log", &stat.name))
        .map(|stat| {
            let fname = parent
                .join(stat.fhash)
                .join(format!("{:02x}", stat.id))
                .join(&stat.name);
            fname.display().to_string()
        })
        .for_each(|filename| {
            operator_script_excute(lib_path_name, &filename, args);
        });

    Ok(())
}

/**
任务接收API，服务器接到并行执行任务后，通过tx提交给worker执行。
 - body是Post提交的，是传递给算子的json参数数据，数据对当前系统透明。
 - body将在worker中提取 PARALLEL_RECORD_KEY 数据，用于记录计算结果；
 - 当前服务器只负责针对指定(min,max)范围的ihash节点下文件进行计算，不了解全局的分配情况，所以通过借助router，可以充分调用所有的服务器进行并行计算。
 */
pub async fn gfs_parallel_task(
    Query(para): Query<ParallelArgs>,
    State(shared): State<SharedHandle>,
    body: Bytes,
) -> AResult<Response, MyError> {
    event!(target: "parallel", loglevel(Debug), "para = {:#?}", para);

    if para.min < 0 || para.max > 255 || para.min >= para.max {
        return Err(MyError::AnyErr(anyhow!(
            "min,max is error, should in [0,255]"
        )));
    }
    // min, max 用于控制或指定计算范围，也将用于作为记录结果的等级index。
    if let Ok(args) = std::str::from_utf8(&body) {
        shared
            .ptx
            .send(ParaTask::Parallel(
                para.libname,
                para.idx,
                para.min,
                para.max,
                args.to_string(),
            ))
            .await
            .or_else(|_| bail!("error to send"))?;
    } else {
        return Err(MyError::AnyErr(anyhow!("args is error: invalidate char")));
    }
    Ok("OK".into_response())
}

/**
读取指定算子（名称）的任务执行情况（只能从被指定为PARALLEL_RECORD_KEY 的服务器读取 ）
 - 数据的含义可以参加TaskStatus的定义（U,E,N,O）。
 - 虽然gfs中任何一个角色的服务器都可以承担这个角色任务，但最好还是选择某个Node的master更合适。
 - 此服务没有容灾备份，只有指定的一台，是单点。
 */
pub async fn gfs_parallel_status(
    Path(task): Path<String>,
    Query(mode): Query<HashMap<String, String>>,
    State(shared): State<SharedHandle>,
) -> AResult<Response, MyError> {
    event!(target: "parallel", loglevel(Debug), "mode = {:#?}, task = {}", mode, task );

    let mut html = false;
    if let Some(mode) = mode.get("mode") {
        if mode.eq("html") {
            html = true;
        }
    }

    let stat = shared.record.read().await;
    if html {
        let hs = stat.read_status_html(&task)?;
        let mut src = String::new();
        src.push_str("<html><body>");
        src.push_str(&hs);
        src.push_str("</body></html>");
        return Ok(Html(src).into_response());
    } else {
        let ret = stat.read_status(&task)?;
        Ok(ret.into_response())
    }
}

// extractor that shows how to consume the request body upfront

const TASKSTATUS_MAX_LEN: usize = 10;
pub const TASK_STATUS_DONE: char = 'O';
pub const TASK_STATUS_ERRO: char = 'E';
pub const TASK_STATUS_NONE: char = 'N';
pub const TASK_STATUS_LOST: char = 'U';
/**
用于记录分布式任务的各个节点的执行情况
 - stat中Value为256个节点的状态数据，key为算子名称。
 - O表示OK,E表示Error，N表示无数据，U表示没有任务执行。
 - timeout用于缓存过期清理，默认48小时有效。

 */
pub struct TaskStatus {
    /// key=operator_name, value is (start_time, idx)
    //timeout: HashMap<String, (std::time::SystemTime, i32)>,
    /// U, E, O, N
    stat: HashMap<String, Vec<char>>,
    fifo: Vec<String>,
}

impl TaskStatus {
    pub(crate) fn new() -> Self {
        Self {
            fifo: vec![],
            stat: HashMap::new(),
        }
    }
    pub(crate) fn set_status(&mut self, key: &String, option: char, ihash: usize) {
        if self.stat.contains_key(key) {
            if let Some(stat) = self.stat.get_mut(key) {
                stat[ihash % 256] = option;
            }
        } else {
            error!(
                "no key = {:#?}, create new !!, exist = {}",
                &key,
                self.stat.len()
            );
            // self.cache_check();
            let mut stat = vec![TASK_STATUS_LOST; 256];
            stat[ihash % 256] = option;
            self.stat.insert(key.clone(), stat);
            self.fifo.push(key.clone());
        }
    }

    pub(crate) fn cache_check(&mut self) {
        while self.stat.len() > TASKSTATUS_MAX_LEN {
            let k = self.fifo.first();
            if let Some(k) = k {
                self.stat.remove(k);
                self.fifo.remove(0);
            }
        }
    }

    pub(crate) fn remove(&mut self, key: &str) {
        let mut ridx = self.fifo.len() + 1;
        for i in 0..self.fifo.len() {
            if self.fifo[i].as_str().eq(key) {
                ridx = i;
                break;
            }
        }
        if ridx < self.fifo.len() {
            self.stat.remove(key);
            self.fifo.remove(ridx);
        }
    }

    pub(crate) fn read_status(&self, key: &String) -> Result<String> {
        event!(target: "parallel", loglevel(Debug), "current are {:#?}", self.fifo);

        if self.stat.contains_key(key) {
            if let Some(stat) = self.stat.get(key) {
                let mut ret = String::new();
                for k in 0..stat.len() {
                    if k % 16 == 0 {
                        ret.push_str("\n");
                    } else {
                        ret.push_str(", ");
                    }
                    ret.push_str(format!("{:02x}:{}", k, stat[k]).as_str());
                }
                return Ok(ret);
            }
        }
        bail!("no such task")
    }

    pub(crate) fn read_status_html(&self, key: &String) -> Result<(String)> {
        if self.stat.contains_key(key) {
            if let Some(stat) = self.stat.get(key) {
                let mut ret = String::new();
                //let id = 0;
                ret.push_str("<div><table>");
                for k in 0..stat.len() {
                    if k % 16 == 0 {
                        ret.push_str("<tr>");
                    }
                    if stat[k] == 'E' {
                        ret.push_str(
                            format!(
                                "<td>{:02x}:<span style='color:red'>{}</span></td>",
                                k, stat[k]
                            )
                            .as_str(),
                        );
                    } else if stat[k] == 'O' {
                        ret.push_str(
                            format!(
                                "<td>{:02x}:<span style='color:green'>{}</span></td>",
                                k, stat[k]
                            )
                            .as_str(),
                        );
                    } else if stat[k] == 'U' {
                        ret.push_str(
                            format!(
                                "<td>{:02x}:<span style='color:gray'>{}</span></td>",
                                k, stat[k]
                            )
                            .as_str(),
                        );
                    } else {
                        ret.push_str(format!("<td>{:02x}:{}</td>", k, stat[k]).as_str());
                    }
                    if k % 16 == 15 {
                        ret.push_str("</tr>");
                    }
                }
                ret.push_str("</table></div>");
                return Ok((ret));
            }
        }
        bail!("no such task")
    }
}

#[test]

fn task_grep_test() {
    let files = vec!["hello.log", "test.log", "abc.logs", "log.txt"];
    let patterns = vec!["*.log", "(*).log", ".log", "(.*).log"];
    for p in patterns.iter() {
        for f in files.iter() {
            if parallel_task_filter(p, f) {
                println!("Yes: {}, {}", p, f);
            } else {
                error!("None: {}, {}", p, f);
            }
        }
    }

    let id = 5140;
    println!("5140 -> 16 -> {:02x}", id);
}

#[allow(dead_code)]
pub enum ParaTask {
    /**
    并行计算指令, Parallel(libname, idx, min, max, json_body)
     */
    Parallel(String, i32, i32, i32, String),
}

impl ParaTask {
    /**
    接收到的指令，执行相应的任务
     * 这里通过 match ... ref mut 进行了‘可变对象‘解构
     * 需要提醒的是，run传递的参数不能是run(&self...)，而应该 run(&mut self...)
     */
    pub async fn run(&mut self, shared: &ShareComm) -> Result<()> {
        match self {
            ParaTask::Parallel(ref mut operator_name, idx, min, max, json_body) => {
                event!(target: "parallel", loglevel(Debug), "Parallel: {}, {}", &operator_name, *idx );

                let ihash = string_hash256(&operator_name);
                let libso =
                    gfs_hashcode_to_path(&shared.gfs.root, ihash, *idx).join(&operator_name);
                if !libso.exists() {
                    bail!(format!(
                        "operator file {:#?} is not exist!",
                        libso.display()
                    ))
                }

                let record_filename = gfsc_parallel_record_name(&operator_name);
                let args: Value = serde_json::from_str(&json_body)?;
                // let writer = args[PARALLEL_RECORD_KEY].as_str().unwrap_or_default();
                // let src = String::from_utf8_lossy(&json_body);
                // let v: Value = serde_json::from_str(&src)?;
                let writer = if args[PARALLEL_RECORD_KEY] != json!(null) {
                    let writer = args[PARALLEL_RECORD_KEY].as_str().unwrap_or_default();
                    writer
                } else {
                    error!("no record_server in body");
                    ""
                };

                let record_idx = args[PARALLEL_RECORD_IDX].as_str();
                let record_idx = if let Some(record_idx) = record_idx {
                    record_idx.parse::<i32>()?
                } else {
                    error!("mybe error, should translate record_idx");
                    *idx
                };

                let taskname = operator_name.clone();
                event!(target: "parallel", loglevel(Debug), "writer = {}, record_fname = {}, task={}",writer, &record_filename, taskname);

                let hash_min = *min;
                let hash_max = *max + 1;
                for i_hash in hash_min..hash_max {
                    event!(target: "parallel", loglevel(Debug), " scan ihash = {}", i_hash );

                    let files = shared.gfs.get_hash_files(i_hash).await?;
                    if files.len() > 0 {
                        let _sc = call_operator_script(
                            &libso.display().to_string(),
                            json_body,
                            files,
                            &shared.gfs.root,
                        )?;
                    }
                    if writer.len() > 0 {
                        let body = bytes::Bytes::from("");
                        gfsc_server_record(
                            writer,
                            ihash,
                            record_idx,
                            &record_filename,
                            RECORD_CMD_DONE,
                            &taskname,
                            i_hash,
                            body,
                        )
                        .await?
                    }
                }
            }
            //  ParaTask(Shutdown()) => { break  },
            //_ => bail!("todo"),
        }
        Ok(())
    }
}
